package canal;

import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.sink.exception.CanalSinkException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

/**
 * 利用canal解析本地binlog
 * https://blog.csdn.net/weixin_40803011/article/details/121889139
 */
public class LocalBinlogEventParserTest {

    private static final Logger dataChangeLog = LoggerFactory.getLogger(LocalBinlogEventParserTest.class);

    private static final String MYSQL_ADDRESS = "127.0.0.1";
    // private static final String USERNAME = "canal";
    private static final String USERNAME = "root";
    // private static final String PASSWORD = "canal";
    private static final String PASSWORD = "123456";
    // 需要提前在该目录放好binlog文件
    private static final String directory = "D:\\work\\临时仓库";

    public static void main(String[] args) throws Exception {
        LocalBinlogEventParserTest parserTest = new LocalBinlogEventParserTest();
        parserTest.test();
    }

    public void test() throws InterruptedException {
        // 在线时间戳转换 https://www.beijing-time.org/shijianchuo/
        // 1638543000000L 2021-12-3 22:50:0
        final EntryPosition defaultPosition = buildPosition("mysql-bin.000114", null, 1638543000000L);
        final LocalBinlogEventParser controller = new LocalBinlogEventParser();
        controller.setMasterPosition(defaultPosition);
        controller.setMasterInfo(buildAuthentication());
        controller.setDirectory(directory);
        controller.setEventSink(new CanalEventSink<List<CanalEntry.Entry>>() {
            @Override
            public void start() {

            }

            @Override
            public void stop() {

            }

            @Override
            public boolean isStart() {
                return false;
            }

            @Override
            public boolean sink(List<CanalEntry.Entry> entries, InetSocketAddress inetSocketAddress, String s) throws CanalSinkException, InterruptedException {
                logRecordEntry(entries);
                return true;
            }

            @Override
            public void interrupt() {

            }
        });
        // 过滤（只监听 库名.表名 的变化）
        controller.setEventFilter(new AviaterRegexFilter("dbname.tablename"));
        // 设置自定义position管理
        controller.setLogPositionManager(new AbstractLogPositionManager() {

            public void persistLogPosition(String destination, LogPosition logPosition) {
//                System.out.println(logPosition);
            }

            @Override
            public LogPosition getLatestIndexBy(String destination) {
                return null;
            }
        });
        controller.start();

        CountDownLatch countDownLatch = new CountDownLatch(1);
        countDownLatch.await();
    }

    private EntryPosition buildPosition(String binlogFile, Long offest, Long timestamp) {
        return new EntryPosition(binlogFile, offest, timestamp);
    }

    private AuthenticationInfo buildAuthentication() {
        return new AuthenticationInfo(new InetSocketAddress(MYSQL_ADDRESS, 3306), USERNAME, PASSWORD);
    }

    private void logRecordEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:"
                            + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChange.getEventType();
                String tableName = entry.getHeader().getTableName();
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    if (eventType == CanalEntry.EventType.DELETE) {
                        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                        Map<String, String> columnDataMap = getColumnDataMap(beforeColumnsList);
                        dataChangeLog.info("删除{}表记录前的数据:{}", tableName, JSON.toJSON(columnDataMap));
                    } else if (eventType == CanalEntry.EventType.INSERT) {
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        dataChangeLog.info("新增{}表的数据", tableName);
                    } else {
                        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        dataChangeLog.info("更新{}表的变更数据", tableName);
                    }
                }
            }
        }
    }

    private Map<String, String> getColumnDataMap(List<CanalEntry.Column> beforeColumnsList) {
        Map<String, String> columnDataMap = new LinkedHashMap<>();
        for (CanalEntry.Column column : beforeColumnsList) {
            columnDataMap.put(column.getName(), column.getValue());
        }
        return columnDataMap;
    }

}
